package com.sunzm.flink.loginfail

import org.apache.commons.lang3.StringUtils
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 * 使用状态编程对2秒内连续登录失败3次的用户进行告警
 */
object StateLoginFailCheckJob {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    val failCOuntOver3OutputTag = new OutputTag[String]("failCOuntOver3OutputTag")

    //为了便于观察，并行度设置为1，实际生产环境不能设置成1
    env.setParallelism(1)
    //读取数据
    //用户id,时间,类型（0成功,1失败）
    //u1001,20210704175816001,0
    val dataStream: DataStream[UserAction] = env.socketTextStream("82.156.210.70", 9999)
      .filter(!_.isEmpty)
      .map(line => {
        val fields = StringUtils.split(line, ",")
        UserAction(fields(0), fields(1), fields(2).toInt)
      })

    val resultDS: DataStream[String] = dataStream.keyBy(_.userId)
      .process(new KeyedProcessFunction[String, UserAction, String] {

        /**
         * 记录每个用户登录失败次数的状态
         */
        private var failCountState: ValueState[Int] = _
        /**
         * 记录有没有发送过告警信息的状态(0没发送过，1已发送过)
         */
        private var sendFlagState: ValueState[Int] = _
        /**
         * 记录第一次登录失败的时间的状态
         */
        private var firstFailTimeState: ValueState[Long] = _


        override def open(parameters: Configuration): Unit = {
          failCountState = getRuntimeContext.getState(
            new ValueStateDescriptor[Int]("failCountState", classOf[Int])
          )

          sendFlagState = getRuntimeContext.getState(
            new ValueStateDescriptor[Int]("sendFlagState", classOf[Int])
          )

          firstFailTimeState = getRuntimeContext.getState(
            new ValueStateDescriptor[Long]("firstFailTimeState", classOf[Long])
          )
        }

        override def processElement(value: UserAction,
                                    ctx: KeyedProcessFunction[String, UserAction, String]#Context,
                                    out: Collector[String]): Unit = {
          //获取历史状态中记录的登录失败的次数
          val oldFailCount = failCountState.value()
          val sendFlag = sendFlagState.value()

          //计算新的登录失败次数
          var newFailCount = oldFailCount

          val userID = ctx.getCurrentKey
          logger.info(s"${userID}, 历史登录失败次数: ${oldFailCount}")

          if (value.actionType == 1) {
            // 登录失败。失败次数+1
            newFailCount += 1

            if(sendFlag == 0){
              //如果已经发送过告警信息，就不需要更新状态了
              //更新状态
              failCountState.update(newFailCount)
            }

            if (oldFailCount == 0) {
              //说明是第一次登录失败
              //记录第一次登录失败的时间
              val firstFailTime = System.currentTimeMillis()
              firstFailTimeState.update(firstFailTime)
              //注册一个2秒的定时器
              ctx.timerService().registerProcessingTimeTimer(firstFailTime + 2000L)
            }
          }

          if (newFailCount >= 3) {
            if (sendFlag == 0) {
              //连续登录失败3次,并且没有发送过告警信息，就告警
              ctx.output(failCOuntOver3OutputTag, s"${userID} 2秒内连续登录失败超过3次，请关注...")

              //更新状态
              sendFlagState.update(1)

              //清空定时器
              ctx.timerService().deleteProcessingTimeTimer(firstFailTimeState.value() + 2000L)
            }
          } else {
            out.collect(s"${value.userId}, ${value.dateTimeStr}, ${value.actionType}")
          }

        }


        override def onTimer(timestamp: Long,
                             ctx: KeyedProcessFunction[String, UserAction, String]#OnTimerContext,
                             out: Collector[String]): Unit = {

          //如果2秒内登录失败没有超过3次，就会触发定时器，清空所有状态
          failCountState.clear()
          firstFailTimeState.clear()
          sendFlagState.clear()

        }

        override def close(): Unit = super.close()
      })

    resultDS.print("主流")

    //获取侧输出流中的数据
    resultDS.getSideOutput(failCOuntOver3OutputTag).print("侧流")

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }

  private case class UserAction(
                                 /**
                                  * 用户ID
                                  */
                                 userId: String,

                                 /**
                                  * 操作时间 yyyyMMddHHmmssSSS
                                  */
                                 dateTimeStr: String,

                                 /**
                                  * 操作类型 0成功，1失败
                                  */
                                 actionType: Int)

}
